package com.itcode.Deserializer;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

public class CustomerDeserializationSchema implements DebeziumDeserializationSchema<String> {

    private static final String DATABASE = "database";
    private static final String TABLE_NAME = "tableName";
    private static final String BEFORE = "before";
    private static final String AFTER = "after";

    private static final String TYPE = "type";

    private static final String DOT_SEPARATOR = "\\.";

    /**
     * 封装的格式
     * {
     * "database":"",
     * "tableName":"",
     * "type":"c,r,u,d",
     * "before":{},
     * "after":{}
     * }
     */
    @Override
    public void deserialize(SourceRecord record, Collector<String> out) throws Exception {
        JSONObject jsonObject = new JSONObject();
        //1.封装数据库和表名
        String topic = record.topic();
        String[] topicAttributes = topic.split(DOT_SEPARATOR);
        jsonObject.put(DATABASE, topicAttributes[1]);
        jsonObject.put(TABLE_NAME, topicAttributes[2]);
        //2.封装操作类型
        Envelope.Operation operation = Envelope.operationFor(record);
        String operator = operation.name();
        jsonObject.put(TYPE, "create".equalsIgnoreCase(operator) ? "insert" : operator.toLowerCase());
        //3.封装before和after
        Struct value = (Struct) record.value();
        extractFields(jsonObject, value, BEFORE);
        extractFields(jsonObject, value, AFTER);
        out.collect(jsonObject.toJSONString());
    }

    private void extractFields(JSONObject jsonObject, Struct value, String flag) {
        Struct flagVal = value.getStruct(flag);
        JSONObject flagJson = new JSONObject();
        if (flagVal != null) {
            for (Field field : flagVal.schema().fields()) {
                //注意不能使用toString
                flagJson.put(field.name(), flagVal.get(field));
            }
        }
        jsonObject.put(flag, flagJson.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
